package com.hainiu.cat.web.kafka.producer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * create by biji.zhao on 2021/6/29
 */
@RestController
public class KafkaProducer {

//    @Resource
//    private KafkaTemplate<String, Object> kafkaTemplate;
//
//    @GetMapping("/kafka/normal/{message}")
//    public void sendMessage1(@PathVariable("message") String normalMessage) {
//        kafkaTemplate.send("topic_filter", normalMessage);
//    }
//
//    @GetMapping("/kafka/callbackOne/{message}")
//    public void sendCallbackOne(@PathVariable("message") String normalMessage) {
//        kafkaTemplate.send("topic1", normalMessage)
//                .addCallback(success -> {
//                    // 消息发送到的topic
//                    String topic = success.getRecordMetadata().topic();
//                    // 消息发送到的分区
//                    int partition = success.getRecordMetadata().partition();
//                    // 消息在分区内的offset
//                    long offset = success.getRecordMetadata().offset();
//                    System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
//                }, failure -> {
//                    System.out.println("发送消息失败:" + failure.getMessage());
//                });
//    }
//
//    @GetMapping("/kafka/callbackTwo/{message}")
//    public void sendMessage3(@PathVariable("message") String callbackMessage) {
//        kafkaTemplate.send("topic_callback", callbackMessage)
//                .addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
//                    @Override
//                    public void onFailure(Throwable ex) {
//                        System.out.println("发送消息失败：" + ex.getMessage());
//                    }
//                    @Override
//                    public void onSuccess(SendResult<String, Object> result) {
//                        System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
//                                + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
//                    }
//                });
//    }
//
//    @GetMapping("/kafka/transaction")
//    public void sendTransaction() {
//        // 声明事务：后面报错消息不会发出去
//        kafkaTemplate.executeInTransaction(operations -> {
//            operations.send("topic_1", "test executeInTransaction");
//            throw new RuntimeException("fail");
//        });
//        // 不声明事务：后面报错但前面消息已经发送成功了
//        kafkaTemplate.send("topic_1", "test executeInTransaction");
//        throw new RuntimeException("fail");
//    }
//
//
//    /**
//     * 从topic1接收到的消息经过处理后转发到topic2
//     **/
//    @KafkaListener(topics = {"topic1"})
//    @SendTo("topic2")
//    public String onMessage7(ConsumerRecord<?, ?> record) {
//        return record.value()+"-forward message";
//    }
}
